package demo.db.zookeeper;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkUtils {

    static Logger logger = LoggerFactory.getLogger(ZkUtils.class);
    
    public static ZooKeeper buildZkClient(String connectString) throws IOException, InterruptedException {
        return buildZkClient(connectString, 5000, false);
    }

    public static ZooKeeper buildZkClient(String connectString, int sessionTimeout)
            throws IOException, InterruptedException {
        return buildZkClient(connectString, sessionTimeout, false);
    }

    /**
     * To create a ZooKeeper client object, the application needs to pass a connection string containing
     * a comma separated list of host:port pairs, each corresponding to a ZooKeeper server.
     * <p>
     * Session establishment is asynchronous. This constructor will initiate connection to the server
     * and return immediately - potentially (usually) before the session is fully established. The
     * watcher argument specifies the watcher that will be notified of any changes in state. This
     * notification can come at any point before or after the constructor call has returned.
     * <p>
     * The instantiated ZooKeeper client object will pick an arbitrary server from the connectString and
     * attempt to connect to it. If establishment of the connection fails, another server in the connect
     * string will be tried (the order is non-deterministic, as we random shuffle the list), until a
     * connection is established. The client will continue attempts until the session is explicitly
     * closed.
     * <p>
     * Added in 3.2.0: An optional "chroot" suffix may also be appended to the connection string. This
     * will run the client commands while interpreting all paths relative to this root (similar to the
     * unix chroot command).
     *
     * @param connectString
     *            comma separated host:port pairs, each corresponding to a zk server. e.g.
     *            "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002" If the optional chroot suffix is used
     *            the example would look like: "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a"
     *            where the client would be rooted at "/app/a" and all paths would be relative to this
     *            root - ie getting/setting/etc... "/foo/bar" would result in operations being run on
     *            "/app/a/foo/bar" (from the server perspective).
     * @param sessionTimeout
     *            session timeout in milliseconds
     * @param watcher
     *            a watcher object which will be notified of state changes, may also be notified for
     *            node events
     * @param canBeReadOnly
     *            (added in 3.4) whether the created client is allowed to go to read-only mode in case
     *            of partitioning. Read-only mode basically means that if the client can't find any
     *            majority servers but there's partitioned server it could reach, it connects to one in
     *            read-only mode, i.e. read requests are allowed while write requests are not. It
     *            continues seeking for majority in the background.
     *
     * @throws IOException
     *             in cases of network failure
     * @throws IllegalArgumentException
     *             if an invalid chroot path is specified
     */
    public static ZooKeeper buildZkClient(String connectString, int sessionTimeout, boolean canBeReadOnly)
            throws IOException, InterruptedException {
        CountDownLatch connectedSignal = new CountDownLatch(1);
        ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if (watchedEvent == null || watchedEvent.getType() == null) {
                    return;
                }
                logger.info("触发{}事件", watchedEvent);
                if (watchedEvent.getState() == KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        }, canBeReadOnly);

        connectedSignal.await();
        return zooKeeper;
    }
}
